package app.dwd;

import app.ods.Constant;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;

/**
 * ODS层：用户行为原始数据生成器（模拟业务系统数据采集）
 */
public class UserBehaviorKafkaProducer {
    private static final Random random = new Random();
    // 模拟用户ID（10001-10010）
    private static final String[] USER_IDS = {"10001", "10002", "10003", "10004", "10005", "10006", "10007", "10008", "10009", "10010"};
    // 模拟商品ID（关联后续商品属性表）
    private static final String[] GOODS_IDS = {"G001", "G002", "G003", "G004", "G005", "G006", "G007", "G008", "G009", "G010"};
    // 模拟行为类型
    private static final String[] BEHAVIOR_TYPES = {
            Constant.BEHAVIOR_TYPE_PURCHASE,
            Constant.BEHAVIOR_TYPE_SEARCH,
            Constant.BEHAVIOR_TYPE_COLLECT,
            Constant.BEHAVIOR_TYPE_BROWSE
    };

    public static void main(String[] args) {
        // 1. 配置Kafka生产者
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-node1:9092,kafka-node2:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "1");

        // 2. 创建生产者实例
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            // 3. 模拟生成1000条原始数据（实际场景为业务系统实时采集）
            for (int i = 0; i < 1000; i++) {
                String userId = USER_IDS[random.nextInt(USER_IDS.length)];
                String goodsId = GOODS_IDS[random.nextInt(GOODS_IDS.length)];
                String behaviorType = BEHAVIOR_TYPES[random.nextInt(BEHAVIOR_TYPES.length)];
                // 模拟行为时间（近90天内的随机时间）
                long currentTime = System.currentTimeMillis();
                long behaviorTime = currentTime - random.nextInt(90 * 24 * 60 * 60 * 1000);
                // 搜索行为特殊处理：添加搜索关键词
                String searchKeyword = (Constant.BEHAVIOR_TYPE_SEARCH.equals(behaviorType)) 
                        ? (random.nextBoolean() ? "男童T恤" : "女童连衣裙") 
                        : "";

                // 构建JSON格式原始数据（实际场景建议用FastJSON/Jackson序列化）
                String value = String.format(
                        "{\"user_id\":\"%s\",\"goods_id\":\"%s\",\"behavior_type\":\"%s\",\"behavior_time\":%d,\"search_keyword\":\"%s\"}",
                        userId, goodsId, behaviorType, behaviorTime, searchKeyword
                );

                // 发送数据到Kafka ODS主题
                ProducerRecord<String, String> record = new ProducerRecord<>(
                        Constant.ODS_USER_BEHAVIOR_TOPIC,
                        userId, // 按用户ID分区，保证同一用户数据有序
                        value
                );
                producer.send(record);
                Thread.sleep(100); // 模拟数据产生间隔
            }
            System.out.println("ODS层原始数据发送完成");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}